//package com.china.produce.listener;
//
//import com.alibaba.fastjson.JSONObject;
//import com.china.produce.pojo.KafkaConsumeThreadAttr;
//import lombok.extern.slf4j.Slf4j;
//import org.apache.commons.compress.utils.Lists;
//import org.apache.kafka.clients.consumer.ConsumerRecord;
//import org.springframework.beans.factory.InitializingBean;
//import org.springframework.beans.factory.annotation.Autowired;
//import org.springframework.beans.factory.annotation.Qualifier;
//import org.springframework.data.redis.core.RedisTemplate;
//import org.springframework.kafka.annotation.KafkaListener;
//import org.springframework.kafka.support.Acknowledgment;
//import org.springframework.stereotype.Component;
//import java.util.ArrayList;
//import java.util.Iterator;
//import java.util.Map;
//import java.util.concurrent.ExecutorService;
//import java.util.concurrent.Executors;
//import java.util.concurrent.ThreadFactory;
//
///**
// * @ClassName
// * @Author niujie
// * @Version
// * @Description
// * @CreateTime 2023/8/5
// */
//@Component
//@Slf4j
//public class KafkaMultiConsumerListener implements InitializingBean {
//
//
//    @Autowired
//    @Qualifier("redisTemplate2")
//    private RedisTemplate<String, Object> redisTemplate;
//
//    private ArrayList<KafkaConsumeThreadAttr> threads = Lists.newArrayList();
//
//    @KafkaListener(topics = {"test_topic"})
//    public void batchConsume(ConsumerRecord<?, ?> consumerRecord, Acknowledgment ack) {
//        ack.acknowledge();
//        //获取消费的数据，有关闭的不再执行下边的逻辑
//        String value = (String) consumerRecord.value();
//        Map map = JSONObject.parseObject(value, Map.class);
//        String switches = (String) map.get("switches");
//        if (!"open".equals(switches)) {
//            return;
//        }
//        Integer tabId = (Integer) map.get("tabId");
//        //怎么判断当前线程ID和userid存在的映射关系,假设最多可以开3个线程消费
//        String tabKey = "tab-" + tabId % 3;
//        String state = (String) redisTemplate.opsForValue().get(tabKey);
//        if ("close".equals(state)) {
//            toStopTabId(tabId);
//            redisTemplate.opsForValue().set(tabKey, "closed");
//        }
//        if ("running".equals(state)) {
//            toRunTabId(tabId);
//            redisTemplate.opsForValue().set(tabKey, "closed");
//        }
//        //判断线程中有没有和执行试验线对齐的
//        threads.stream().forEach(threadPojo -> {
//            if (tabKey.equals(threadPojo.getName()) && !threadPojo.getExecutor().isShutdown()) {
//                //已经有线程在执行，直接获取
//                threadPojo.getExecutor().execute(new TabRunnable(value));
//                return;
//            }
//        });
//    }
//
//    /**
//     * 停止具体的实验项目
//     *
//     * @param tabId
//     */
//    public void toStopTabId(Integer tabId) {
//        String tabKey = "tab-" + tabId % 3;
//        String tabName = "tab-name-" + tabId % 3;
//
//        log.info("stop tabKey {}", tabKey);
//        threads.stream().forEach(threadAttr -> {
//            if (tabKey.equals(threadAttr.getName())) {
//                ExecutorService executor = threadAttr.getExecutor();
//                executor.shutdown();
//                redisTemplate.opsForValue().set(tabName, "stop");
//                return;
//            }
//        });
//    }
//
//    /**
//     * 开始实现项目
//     *
//     * @param tabId
//     */
//    public void toRunTabId(Integer tabId) {
//        //判断线程有没有被停止的
//        String tabName = "tab-name-" + tabId % 3;
//        String tabKey = "tab-" + tabId;
//        String flag = (String) redisTemplate.opsForValue().get(tabName);
//        if (!"stop".equals(flag)) {
//            return;
//        }
//        //删除被停止的线程，重新添加
//        Iterator<KafkaConsumeThreadAttr> iterator = threads.iterator();
//        while (iterator.hasNext()) {
//            KafkaConsumeThreadAttr threadPojo = iterator.next();
//            if (tabKey.equals(threadPojo.getName())) {
//                ExecutorService executor = threadPojo.getExecutor();
//                if(executor.isShutdown()) {
//                    iterator.remove();
//                }
//            }
//        }
//        ExecutorService executor = Executors.newFixedThreadPool(2, new ThreadFactory() {
//            @Override
//            public Thread newThread(Runnable r) {
//                return new Thread(r, tabKey);
//            }
//        });
//
//        KafkaConsumeThreadAttr threadPojo = new KafkaConsumeThreadAttr();
//        threadPojo.setName(tabKey);
//        threadPojo.setExecutor(executor);
//        threads.add(threadPojo);
//        redisTemplate.opsForValue().set(tabName, "running");
//    }
//
//    @Override
//    public void afterPropertiesSet() throws Exception {
//        //初始化创建3个线程池
//        for (int i = 0; i < 3; i++) {
//            String threadName = "tab-" + i;
//            ExecutorService executor = Executors.newFixedThreadPool(2, new ThreadFactory() {
//                @Override
//                public Thread newThread(Runnable r) {
//                    return new Thread(r, threadName);
//                }
//            });
//
//            KafkaConsumeThreadAttr threadAttr = new KafkaConsumeThreadAttr();
//            threadAttr.setName(threadName);
//            threadAttr.setExecutor(executor);
//            threads.add(threadAttr);
//        }
//    }
//
//    class TabRunnable implements Runnable {
//
//        private String msg;
//
//        public TabRunnable(String msg) {
//            this.msg = msg;
//        }
//
//        @Override
//        public void run() {
//            //判断消息中有没有停止的
//            System.out.println("thread name : " + Thread.currentThread().getName() + "thread id : " + Thread.currentThread().getId() + msg);
//
//            //执行消费具体业务
//        }
//    }
//}
